package org.jetlinks.demo.protocol.siterwell.tcp;


import io.netty.buffer.Unpooled;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.DeviceOnlineMessage;
import org.jetlinks.core.message.Message;
import org.jetlinks.core.message.codec.EncodedMessage;
import org.jetlinks.core.message.event.EventMessage;
import org.jetlinks.core.message.function.FunctionInvokeMessage;
import org.jetlinks.core.message.function.FunctionInvokeMessageReply;
import org.jetlinks.core.message.property.ReportPropertyMessage;
import org.jetlinks.core.server.session.DeviceSession;
import org.jetlinks.demo.protocol.siterwell.tcp.message.*;
import org.jetlinks.demo.protocol.siterwell.tcp.util.ByteUtils;
import reactor.core.publisher.Mono;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


@Slf4j
public class TcpMessageHandler {

    private static ConcurrentMap<String, ReadSensorCache> pushMessageCache = new ConcurrentHashMap<>();

    protected Mono<DeviceMessage> dealRequest(DeviceSession session, MessageType messageType, TcpPayload tcpPayload) {

        if (MessageType.DEVICE_ALARM.equals(messageType)) {
            return handleDeviceAlarm(session, (D2SDeviceAlarm) tcpPayload);
        }
        if (MessageType.PING.equals(messageType)) {
            return handlePing(session);
        }
        if (MessageType.AUTH_REQ.equals(messageType)) {
            return handleAuthReq(session, (D2SAuthRequest) tcpPayload);
        }
        if (MessageType.ADD_DEVICE.equals(messageType)) {
            return handleAddDevice(session, (D2SAddDevice) tcpPayload);
        }
        if (MessageType.REMOVE_DEVICE.equals(messageType)) {
            return handleRemoveDevice(session, (D2SRemoveDevice) tcpPayload);
        }
        if (MessageType.RESTORE_LINK.equals(messageType)) {
            return handleRestoreLink(session, (D2SRestoreLink) tcpPayload);
        }
        if (MessageType.POWER_ON.equals(messageType)) {
            return handlePowerOn(session, (D2SPowerOn) tcpPayload);
        }
        if (MessageType.POWER_FAILURE.equals(messageType)) {
            return handlePowerFailure(session, (D2SPowerFailure) tcpPayload);
        }
        if (MessageType.DEPLOY_FOR_DEFENCE.equals(messageType)) {
            return handleDeploy4Defence(session, (D2SDeployForDefence) tcpPayload);
        }
        if (MessageType.DEPLOY_REMOVE.equals(messageType)) {
            return handleDeployRemove(session, (D2SDeployRemove) tcpPayload);
        }
        if (MessageType.VOLTAGE_NORMAL.equals(messageType)) {
            return handleVoltageNormal(session, (D2SVoltageNormal) tcpPayload);
        }
        if (MessageType.NORMAL.equals(messageType)) {
            return handleDeviceStatus(session, (D2SDeviceStatus) tcpPayload);
        }
        if (MessageType.MUTE.equals(messageType)) {
            return handleDeviceStatus(session, (D2SDeviceStatus) tcpPayload);
        }
        if (MessageType.FAULT.equals(messageType)) {
            return handleDeviceStatus(session, (D2SDeviceStatus) tcpPayload);
        }
        if (MessageType.READ_SENSOR.equals(messageType)) {
            return handleReadSensorReply(session, (D2SReadSensorReply) tcpPayload);
        }

        return Mono.empty();
    }

    private Mono<DeviceMessage> handleReadSensorReply(DeviceSession session, D2SReadSensorReply tcpPayload) {

        String gatewayId = tcpPayload.getGateWayId();
        Set<String> keySet = pushMessageCache.keySet();
        if (log.isDebugEnabled()) {
            log.debug("func invoke gatewayId: {} , size: {} ", gatewayId, keySet.size());
        }
        ReadSensorCache cache = pushMessageCache.get(gatewayId);
        String messageId = cache.getMessageId();
        if (log.isDebugEnabled()) {
            log.debug("func invoke gatewayId: {} , messageId: {} ", gatewayId, messageId);
        }
        FunctionInvokeMessageReply reply = new FunctionInvokeMessageReply();
        reply.setFunctionId(cache.getFuncId());
        reply.setOutput(tcpPayload.getSensorDataList());
        reply.setMessageId(messageId);
        reply.setDeviceId(gatewayId);
        return Mono.just(reply);
    }

    private Mono<DeviceMessage> handleDeviceStatus(DeviceSession session, D2SDeviceStatus tcpPayload) {
        String childDeviceId = tcpPayload.getChildDeviceId();

        EventMessage eventMessage = new EventMessage();
        eventMessage.setDeviceId(childDeviceId);
        eventMessage.setEvent("yk_device_status");
        eventMessage.setMessageId(IDGenerator.SNOW_FLAKE_STRING.generate());

        return sendDefaultResponse(session, tcpPayload)
                .doOnNext(resp -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply device report device status tcp message:\n{}", ByteUtils.byte2HexStr(resp));
                    }
                }).thenReturn(eventMessage);

    }

    private Mono<DeviceMessage> handleVoltageNormal(DeviceSession session, D2SVoltageNormal tcpPayload) {
        return reportProperty(session, tcpPayload, "voltage_normal", false);
    }

    private Mono<DeviceMessage> handleDeployRemove(DeviceSession session, D2SDeployRemove tcpPayload) {
        return reportProperty(session, tcpPayload, "deploy_for_defence", false);
    }

    private Mono<DeviceMessage> handleDeploy4Defence(DeviceSession session, D2SDeployForDefence tcpPayload) {
        return reportProperty(session, tcpPayload, "deploy_for_defence", true);
    }

    private Mono<DeviceMessage> handlePowerFailure(DeviceSession session, D2SPowerFailure tcpPayload) {
        return reportProperty(session, tcpPayload, "power_on", false);
    }

    private Mono<DeviceMessage> handlePowerOn(DeviceSession session, D2SPowerOn tcpPayload) {
        return reportProperty(session, tcpPayload, "power_on", true);
    }

    private Mono<DeviceMessage> reportProperty(DeviceSession session, AbstractCommandRequest tcpPayload,
                                               String propertyName, Object value) {
        String gateWayId = tcpPayload.getGateWayId();
        ReportPropertyMessage reportPropertyMessage = new ReportPropertyMessage();
        reportPropertyMessage.setDeviceId(gateWayId);
        reportPropertyMessage.setTimestamp(System.currentTimeMillis());
        Map<String, Object> properties = new HashMap<>();
        properties.put(propertyName, value);
        reportPropertyMessage.setProperties(properties);
        return sendDefaultResponse(session, tcpPayload)
                .doOnNext(resp -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply device report property {} tcp message:\n{}", propertyName, ByteUtils.byte2HexStr(resp));
                    }
                }).thenReturn(reportPropertyMessage);
    }

    private Mono<DeviceMessage> handleRestoreLink(DeviceSession session, D2SRestoreLink tcpPayload) {
        String childDeviceId = tcpPayload.getChildDeviceId();
        DeviceOnlineMessage onlineMessage = new DeviceOnlineMessage();
        onlineMessage.setDeviceId(childDeviceId);
        onlineMessage.setTimestamp(System.currentTimeMillis());
        return sendDefaultResponse(session, tcpPayload)
                .doOnNext(resp -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply device restore link tcp message:\n{}", ByteUtils.byte2HexStr(resp));
                    }
                }).thenReturn(onlineMessage);
    }

    private Mono<DeviceMessage> handleDeviceAlarm(DeviceSession session, D2SDeviceAlarm tcpPayload) {
        String childDeviceId = tcpPayload.getChildDeviceId();

        EventMessage eventMessage = new EventMessage();
        eventMessage.setDeviceId(childDeviceId);
        eventMessage.setEvent("yk_sos");
        eventMessage.setMessageId(IDGenerator.SNOW_FLAKE_STRING.generate());

        return sendDefaultResponse(session, tcpPayload)
                .doOnNext(resp -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply device alarm tcp message:\n{}", ByteUtils.byte2HexStr(resp));
                    }
                }).thenReturn(eventMessage);

    }

    private Mono<DeviceMessage> handleRemoveDevice(DeviceSession session, D2SRemoveDevice tcpPayload) {
        return sendDefaultResponse(session, tcpPayload)
                .doOnNext(resp -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply add device tcp message:\n{}", ByteUtils.byte2HexStr(resp));
                    }
                }).then(Mono.empty());
    }

    /**
     * 处理网关发送到平台的设备增加时的消息
     *
     * @param session
     * @param tcpPayload
     * @return
     */
    private Mono<DeviceMessage> handleAddDevice(DeviceSession session, D2SAddDevice tcpPayload) {

        return sendDefaultResponse(session, tcpPayload)
                .doOnNext(resp -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply add device tcp message:\n{}", ByteUtils.byte2HexStr(resp));
                    }
                }).then(Mono.empty());
    }

    /**
     * 处理网关发送到平台的心跳消息
     *
     * @param session
     * @return
     */
    private Mono<DeviceMessage> handlePing(DeviceSession session) {
        byte[] responseBytes = new S2DPong().toBytes();
        return session
                .send(EncodedMessage.simple(Unpooled.wrappedBuffer(responseBytes)))
                .doOnNext(success -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply ping tcp message:\n{}", ByteUtils.byte2HexStr(responseBytes));
                    }
                }).then(Mono.empty());
    }

    /**
     * 处理网关发送到平台的登录请求的消息
     *
     * @param session
     * @param authRequest
     * @return
     */
    private Mono<DeviceMessage> handleAuthReq(DeviceSession session, D2SAuthRequest authRequest) {

        String gateWayId = authRequest.getGateWayId();
        DeviceOnlineMessage onlineMessage = new DeviceOnlineMessage();
        Map<String, Object> headers = new HashMap<>();
        headers.put("keepOnlineTimeoutSeconds", 240);
        onlineMessage.setHeaders(headers);
        onlineMessage.setDeviceId(gateWayId);
        onlineMessage.setTimestamp(System.currentTimeMillis());
        byte[] responseBytes = S2DAuthResponse.of(authRequest).toBytes();
        return session
                .send(EncodedMessage.simple(Unpooled.wrappedBuffer(responseBytes)))
                .doOnNext(success -> {
                    if (log.isDebugEnabled()) {
                        log.debug("reply tcp message:\n{}", ByteUtils.byte2HexStr(responseBytes));
                    }
                })
                .thenReturn(onlineMessage);
    }

    private Mono<byte[]> sendDefaultResponse(DeviceSession session, AbstractCommandRequest request) {
        byte[] responseBytes = S2DDefaultReplyResponse.of(request).toBytes();
        return session
                .send(EncodedMessage.simple(Unpooled.wrappedBuffer(responseBytes)))
                .thenReturn(responseBytes);
    }


    protected Mono<EncodedMessage> pushMessage(DeviceSession session, Message message) {
        EncodedMessage encodedMessage = null;

        // 功能调用
        if (message instanceof FunctionInvokeMessage) {
            FunctionInvokeMessage functionInvokeMessage = (FunctionInvokeMessage) message;
            String messageId = functionInvokeMessage.getMessageId();
            String funcId = functionInvokeMessage.getFunctionId();
            String deviceId = functionInvokeMessage.getDeviceId();

            if ("func_read_sensor".equals(funcId)) {
                S2DReadSensor s2DReadSensor = new S2DReadSensor(functionInvokeMessage);
                byte[] bytes = s2DReadSensor.toBytes();
                encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(bytes));
                ReadSensorCache readSensorCache = new ReadSensorCache(funcId, messageId, System.currentTimeMillis());
                pushMessageCache.put(deviceId, readSensorCache);
                Set<String> keySet = pushMessageCache.keySet();
                if (log.isDebugEnabled()) {
                    log.debug("invoke func <func_read_sensor> gatewayId: {} , messageId: {} , size: {} , tcp message:\n{}", deviceId, messageId, keySet.size(), ByteUtils.byte2HexStr(bytes));
                }
                return Mono.just(encodedMessage);
            }

            if("func_write_sensor".equals(funcId)){
                S2DWriteSensor s2DReadSensor = new S2DWriteSensor(functionInvokeMessage);
                if (log.isDebugEnabled()) {
                    log.debug("invoke func <func_write_sensor> gatewayId: {} , messageId: {}  ", deviceId, messageId );
                }
                return Mono.empty();
            }
        }

        return Mono.empty();
    }

    @AllArgsConstructor
    @Getter
    static class ReadSensorCache {
        private String funcId;
        private String messageId;
        private long readTime;
    }

}
